package com.atguigu.union;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * @author yhm
 * @create 2024-04-03 10:11
 */
public class Test03_Connect {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 并发一定为1  才能实现inner join
        env.setParallelism(1);
        // 2. 读取数据源
        // 主流数据    id,ts
        SingleOutputStreamOperator<WaterSensor> streamOperator1 = env.socketTextStream("hadoop102", 7777).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.parseLong(split[1]), 0);
            }
        });
        // 连接流数据   id,vc
        SingleOutputStreamOperator<WaterSensor> streamOperator2 = env.socketTextStream("hadoop102", 8888).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], 0L, Integer.parseInt(split[1]));
            }
        });

        // 3. 处理数据
        ConnectedStreams<WaterSensor, WaterSensor> connectStream = streamOperator1.connect(streamOperator2);

        SingleOutputStreamOperator<WaterSensor> processStream = connectStream.process(new CoProcessFunction<WaterSensor, WaterSensor, WaterSensor>() {

            // 先到的数据存起来
            // 后到的数据查先到的数据
            HashMap<String, WaterSensor> cache1 ;
            HashMap<String, WaterSensor> cache2 ;

            @Override
            public void open(Configuration parameters) throws Exception {
                // open方法和类初始化一样的   多少并发运行几次
                System.out.println("初始化cache");
                cache1 = new HashMap<String, WaterSensor>();
                cache2 = new HashMap<String, WaterSensor>();
            }

            @Override
            public void processElement1(WaterSensor waterSensor1, Context ctx, Collector<WaterSensor> out) throws Exception {
                System.out.println("调用主流处理");
                // 处理主流数据   id ts
                // 存到cache1中
                String key = waterSensor1.id;
                cache1.put(key, waterSensor1);
                // 查找cache2
                // 如果能join上写出
                if (cache2.containsKey(key)) {
                    WaterSensor waterSensor2 = cache2.get(key);
                    waterSensor1.setVc(waterSensor2.vc);
                    out.collect(waterSensor1);
                }
            }

            @Override
            public void processElement2(WaterSensor waterSensor2, Context ctx, Collector<WaterSensor> out) throws Exception {
                // 处理连接流数据   id vc
                System.out.println("调用连接流处理");
                // 存到cache2中
                String key = waterSensor2.id;
                cache2.put(key, waterSensor2);
                // 查找cache1
                // 如果能join上写出
                if (cache1.containsKey(key)) {
                    WaterSensor waterSensor1 = cache1.get(key);
                    waterSensor2.setTs(waterSensor1.getTs());
                    out.collect(waterSensor2);
                }

            }
        });


        // 4. 输出
        processStream.print();

        // 5. 执行环境
        env.execute();
    }
}
